package com.gizwits.kafkaScala

import java.util.Properties
import java.util.concurrent._

import com.gizwits.util.CaseClasssDomain.User
import com.gizwits.util.MsgCommandDecoder
import kafka.consumer.{Consumer, ConsumerConfig, KafkaStream}
import kafka.serializer.{Decoder, StringDecoder}
import kafka.utils.{Logging, _}

import scala.reflect._
class ScalaConsumerTest(val stream: KafkaStream[String, User], val threadNumber: Int) extends Logging with Runnable {
  def run {
    val it = stream.iterator()
    while (it.hasNext()) {
      val msg = it.next().message()
      System.out.println(System.currentTimeMillis() + ",Thread " + threadNumber + ": " + msg)
    }
    System.out.println("Shutting down Thread: " + threadNumber)
  }
}
object ConsumerScala {
  def createConsumerConfig(zookeeper: String, groupId: String): ConsumerConfig = {
    val props = new Properties()
    props.put("zookeeper.connect", zookeeper)
    props.put("group.id", groupId)
    props.put("auto.offset.reset", "smallest")
    props.put("zookeeper.session.timeout.ms", "400")
    props.put("zookeeper.sync.time.ms", "200")
    props.put("auto.commit.interval.ms", "10")
    val config = new ConsumerConfig(props)
    config
  }
  def   createConsumer(zookeeper: String,groupId: String,topic: String,numThreads:Int)={
    val config = createConsumerConfig(zookeeper, groupId)
    val consumer = Consumer.create(config)
    val topicCountMap = Map(topic -> numThreads)

    // keyDecoder: Decoder[K],
    // valueDecoder: Decoder[V])

    val keyDecoder = classTag[StringDecoder].runtimeClass.getConstructor(classOf[VerifiableProperties])
      .newInstance(config.props)
      .asInstanceOf[Decoder[String]]
    val valueDecoder = classTag[MsgCommandDecoder[User]].runtimeClass.getConstructor(classOf[VerifiableProperties])
      .newInstance(config.props)
      .asInstanceOf[Decoder[User]]

    val consumerMap = consumer.createMessageStreams[String,User](topicCountMap,keyDecoder,valueDecoder)
    val streams = consumerMap.get(topic).get
    val executor: ExecutorService= Executors.newFixedThreadPool(numThreads)
    var threadNumber = 0
    for (stream <- streams) {
      executor.submit(new ScalaConsumerTest(stream, threadNumber))
      threadNumber += 1
    }

  }
  def main(args: Array[String]) {

    createConsumer("s60:2181,s2:2181,s5:2181","1","testkafka-topic",3)



  }
}
